package com.flink.hbase.kafka2elasticsearch;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;

/**
 * JSON 消息转换器
 * 将 Kafka 中的 JSON 消息转换为 UserDocument 对象
 */
public class JsonMessageConverter extends ProcessFunction<String, UserDocument> implements Serializable {
    private static final Logger LOG = LoggerFactory.getLogger(JsonMessageConverter.class);
    private static final long serialVersionUID = 1L;
    
    private transient ObjectMapper objectMapper;
    private static final DateTimeFormatter[] DATE_FORMATTERS = {
        DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"),
        DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss"),
        DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS"),
        DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSX"),
        DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS"),
        DateTimeFormatter.ISO_LOCAL_DATE_TIME
    };
    
    @Override
    public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
        super.open(parameters);
        objectMapper = new ObjectMapper();
        objectMapper.registerModule(new JavaTimeModule());
        LOG.info("JsonMessageConverter initialized");
    }
    
    @Override
    public void processElement(String jsonMessage, Context ctx, Collector<UserDocument> out) throws Exception {
        if (jsonMessage == null || jsonMessage.trim().isEmpty()) {
            LOG.warn("Received empty or null message, skipping");
            return;
        }
        
        try {
            UserDocument document = convertJsonToUserDocument(jsonMessage);
            if (document != null && document.isValid()) {
                out.collect(document);
            } else {
                LOG.warn("Invalid document converted from JSON: {}", jsonMessage);
            }
        } catch (Exception e) {
            LOG.error("Error processing JSON message: {}", jsonMessage, e);
            // 可以选择发送到死信队列或错误处理流
        }
    }
    
    /**
     * 将 JSON 字符串转换为 UserDocument 对象
     */
    private UserDocument convertJsonToUserDocument(String jsonMessage) throws JsonProcessingException {
        JsonNode jsonNode = objectMapper.readTree(jsonMessage);
        
        UserDocument.Builder builder = UserDocument.builder();
        
        // 必需字段
        builder.userId(getStringValue(jsonNode, "user_id", "userId", "id"));
        builder.name(getStringValue(jsonNode, "name", "username", "user_name"));
        
        // 可选字段
        builder.age(getIntegerValue(jsonNode, "age"));
        builder.gender(getStringValue(jsonNode, "gender", "sex"));
        builder.email(getStringValue(jsonNode, "email", "email_address"));
        builder.phone(getStringValue(jsonNode, "phone", "phone_number", "mobile"));
        builder.address(getStringValue(jsonNode, "address"));
        builder.city(getStringValue(jsonNode, "city"));
        builder.country(getStringValue(jsonNode, "country"));
        builder.status(getStringValue(jsonNode, "status", "user_status"));
        builder.tags(getStringValue(jsonNode, "tags", "user_tags"));
        builder.score(getDoubleValue(jsonNode, "score", "user_score"));
        
        // 时间字段
        builder.registrationDate(getDateTimeValue(jsonNode, "registration_date", "reg_date", "created_date"));
        builder.lastLogin(getDateTimeValue(jsonNode, "last_login", "last_login_time"));
        builder.createTime(getDateTimeValue(jsonNode, "create_time", "created_at"));
        builder.updateTime(getDateTimeValue(jsonNode, "update_time", "updated_at"));
        
        // 如果没有创建时间，使用当前时间
        UserDocument document = builder.build();
        if (document.getCreateTime() == null) {
            document.setCreateTime(LocalDateTime.now());
        }
        if (document.getUpdateTime() == null) {
            document.setUpdateTime(LocalDateTime.now());
        }
        
        return document;
    }
    
    /**
     * 从 JSON 节点获取字符串值，支持多个字段名称
     */
    private String getStringValue(JsonNode node, String... fieldNames) {
        for (String fieldName : fieldNames) {
            JsonNode fieldNode = node.get(fieldName);
            if (fieldNode != null && !fieldNode.isNull()) {
                String value = fieldNode.asText();
                if (value != null && !value.trim().isEmpty()) {
                    return value.trim();
                }
            }
        }
        return null;
    }
    
    /**
     * 从 JSON 节点获取整数值
     */
    private Integer getIntegerValue(JsonNode node, String... fieldNames) {
        for (String fieldName : fieldNames) {
            JsonNode fieldNode = node.get(fieldName);
            if (fieldNode != null && !fieldNode.isNull()) {
                if (fieldNode.isInt()) {
                    return fieldNode.asInt();
                } else if (fieldNode.isTextual()) {
                    try {
                        return Integer.parseInt(fieldNode.asText());
                    } catch (NumberFormatException e) {
                        LOG.warn("Invalid integer value for field {}: {}", fieldName, fieldNode.asText());
                    }
                }
            }
        }
        return null;
    }
    
    /**
     * 从 JSON 节点获取双精度值
     */
    private Double getDoubleValue(JsonNode node, String... fieldNames) {
        for (String fieldName : fieldNames) {
            JsonNode fieldNode = node.get(fieldName);
            if (fieldNode != null && !fieldNode.isNull()) {
                if (fieldNode.isNumber()) {
                    return fieldNode.asDouble();
                } else if (fieldNode.isTextual()) {
                    try {
                        return Double.parseDouble(fieldNode.asText());
                    } catch (NumberFormatException e) {
                        LOG.warn("Invalid double value for field {}: {}", fieldName, fieldNode.asText());
                    }
                }
            }
        }
        return null;
    }
    
    /**
     * 从 JSON 节点获取日期时间值
     */
    private LocalDateTime getDateTimeValue(JsonNode node, String... fieldNames) {
        for (String fieldName : fieldNames) {
            JsonNode fieldNode = node.get(fieldName);
            if (fieldNode != null && !fieldNode.isNull()) {
                String dateTimeStr = fieldNode.asText();
                if (dateTimeStr != null && !dateTimeStr.trim().isEmpty()) {
                    return parseDateTime(dateTimeStr.trim());
                }
            }
        }
        return null;
    }
    
    /**
     * 解析日期时间字符串，支持多种格式
     */
    private LocalDateTime parseDateTime(String dateTimeStr) {
        for (DateTimeFormatter formatter : DATE_FORMATTERS) {
            try {
                return LocalDateTime.parse(dateTimeStr, formatter);
            } catch (DateTimeParseException e) {
                // 继续尝试下一个格式
            }
        }
        
        // 尝试解析时间戳
        try {
            long timestamp = Long.parseLong(dateTimeStr);
            // 判断是秒还是毫秒
            if (timestamp > 1000000000000L) {
                // 毫秒时间戳
                return LocalDateTime.ofEpochSecond(timestamp / 1000, 0, java.time.ZoneOffset.UTC);
            } else {
                // 秒时间戳
                return LocalDateTime.ofEpochSecond(timestamp, 0, java.time.ZoneOffset.UTC);
            }
        } catch (NumberFormatException e) {
            LOG.warn("Unable to parse datetime string: {}", dateTimeStr);
        }
        
        return null;
    }
    
    /**
     * 创建带有数据质量验证的转换器
     */
    public static class ValidatingConverter extends JsonMessageConverter {
        
        @Override
        public void processElement(String jsonMessage, Context ctx, Collector<UserDocument> out) throws Exception {
            if (jsonMessage == null || jsonMessage.trim().isEmpty()) {
                LOG.warn("Received empty or null message, skipping");
                return;
            }
            
            try {
                UserDocument document = convertJsonToUserDocument(jsonMessage);
                if (document != null && isHighQualityDocument(document)) {
                    out.collect(document);
                } else {
                    LOG.warn("Low quality document filtered out: {}", jsonMessage);
                }
            } catch (Exception e) {
                LOG.error("Error processing JSON message: {}", jsonMessage, e);
            }
        }
        
        /**
         * 验证文档数据质量
         */
        private boolean isHighQualityDocument(UserDocument document) {
            if (!document.isValid()) {
                return false;
            }
            
            // 检查用户ID格式
            if (document.getUserId().length() < 3) {
                return false;
            }
            
            // 检查姓名长度
            if (document.getName().length() < 2) {
                return false;
            }
            
            // 检查年龄范围
            if (document.getAge() != null && (document.getAge() < 0 || document.getAge() > 150)) {
                return false;
            }
            
            // 检查邮箱格式
            if (document.getEmail() != null && !isValidEmail(document.getEmail())) {
                return false;
            }
            
            return true;
        }
        
        /**
         * 简单的邮箱格式验证
         */
        private boolean isValidEmail(String email) {
            return email.contains("@") && email.contains(".");
        }
    }
    
    /**
     * 工厂方法创建普通转换器
     */
    public static JsonMessageConverter create() {
        return new JsonMessageConverter();
    }
    
    /**
     * 工厂方法创建带验证的转换器
     */
    public static ValidatingConverter createValidating() {
        return new ValidatingConverter();
    }
} 